package com.sink;

import com.Connection.JDBCConnectionTools;
import com.Pojo.NginxLog;
import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.NoArgsConstructor;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Timestamp;


//Flink使用的类都需要实现序列化，进行网络传输
@AllArgsConstructor
public class NginxSink implements Serializable {

    private String sqltype;

    private static final Logger logger = LoggerFactory.getLogger(NginxSink.class);

    public NginxSink() {
    }

    //解析器必须使用FilebeatJSONKeyValueNginxDeserializationSchema才能使用该方法获取sink
    public SinkFunction<String> GetNginxSinkString(String user, String password, String DBip, String databaseName
    ) throws Exception {

        SinkFunction<String> sink = JdbcSink.sink(
                "insert  into NginxLog (time_local,remote_addr,remote_user,request,status,body_bytes_sent,http_referer,connection,upstream_addr,http_user_agent,http_x_forwarded_for," +
                        "url,http_host,request_time,upstream_response_time,upstream_connect_time,upstream_header_time,agent_ip) " +
                        "values" +
                        " (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
                new JdbcStatementBuilder<String>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, String s) throws SQLException {
                        NginxLog log = JSONObject.parseObject(s,NginxLog.class);
                        logger.info(log.toString());   //日志调试-打包需要注释
                        Timestamp localDateTimeToTimeStamp = Timestamp.valueOf(log.getTime_iso8601());
                        preparedStatement.setTimestamp(1, localDateTimeToTimeStamp);    //preparedStatement只接收date和Timestamp，只能将LocalDateTime转为Timestamp再写入DB
                        preparedStatement.setString(2, log.getRemote_addr());
                        preparedStatement.setString(3, log.getRemote_user());
                        preparedStatement.setString(4, log.getRequest().replace("\"",""));
                        preparedStatement.setString(5, log.getStatus());
                        preparedStatement.setString(6, log.getBody_bytes_sent());
                        preparedStatement.setString(7, log.getHttp_referer().replace("\"",""));
                        preparedStatement.setString(8, log.getConnection().replace("\"",""));
                        preparedStatement.setString(9, log.getUpstream_addr());
                        preparedStatement.setString(10, log.getHttp_user_agent().replace("\"",""));
                        preparedStatement.setString(11, log.getHttp_x_forwarded_for().replace("\"",""));
                        preparedStatement.setString(12, log.getUri().replace("\"",""));
                        preparedStatement.setString(13, log.getHttp_host().replace("\"",""));
                        preparedStatement.setString(14, log.getRequest_time());
                        preparedStatement.setString(15, log.getUpstream_response_time());
                        preparedStatement.setString(16, log.getUpstream_connect_time());
                        preparedStatement.setString(17, log.getUpstream_header_time());
                        preparedStatement.setString(18, log.getAgent_ip());
                    }
                },
                JdbcExecutionOptions.builder()
                        .withMaxRetries(3)  //插入发生异常重试次数，Flink checkpoint has started
                        .withBatchSize(1000) //批处理最大条数，设置大一些，生产环境的访问频率高
                        .withBatchIntervalMs(1000) //每次批量写入最小间隔时间
                        .build(),//只要满足其中一条，JDBC批处理就会执行
                new JDBCConnectionTools(this.sqltype).GetConnect(user,password,DBip,databaseName)
        );

        return sink;

    }
}
